package com.chukun.tx.message.queue;

import com.chukun.tx.constants.MessageConstant;
import com.chukun.tx.model.MqMessage;

import java.util.*;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.PriorityBlockingQueue;
import java.util.stream.Collectors;

/**
 * @author chukun
 * 实现简单的时间轮算法
 */
public class TimeWheelQueueProcessor {

    /**
     * 基于错误消息的优先队列
     */
    public static  BlockingQueue<MqMessage> timeQueue;


    public TimeWheelQueueProcessor() {
        timeQueue = new PriorityBlockingQueue<>(8192,
                Comparator.comparing(MqMessage::getNextExpireTime));
    }

    public TimeWheelQueueProcessor(BlockingQueue<MqMessage> timeQueue) {
        this.timeQueue = timeQueue;
    }


    public  synchronized void putFailMessage(MqMessage message) {
        try {
            if (message.getHaveDeadTimes() >= MessageConstant.TIME_INTERVAL.length) {
                // handle by human
                return;
            }
            Map<Long, MqMessage> map = timeQueue.stream()
                    .collect(Collectors.toMap(MqMessage::getMessageId, msg -> msg));
            if (!map.containsKey(message.getMessageId())) {
                timeQueue.put(message);
            }
        }catch (Exception e) {}
    }

    public  MqMessage getFailMessage() {
        try {
            return timeQueue.poll();
        }catch (Exception e) {

        }
        return null;
    }

    public  MqMessage lookTopMsg() {
        try {
            return timeQueue.peek();
        }catch (Exception e) {

        }
        return null;
    }


}
